package com.atguigu.day02;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Example5 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .fromElements("white", "black", "gray")
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String s, Collector<String> collector) throws Exception {
                        if (s.equals("white")) collector.collect(s);
                        else if (s.equals("black")) {
                            collector.collect(s);
                            collector.collect(s);
                        }
                    }
                })
                .print("匿名类");

        env
                .fromElements("white", "black", "gray")
                .flatMap(new MyFlatMap())
                .print("外部类");

        env
                .fromElements("white", "black", "gray")
                .flatMap((String in, Collector<String> out) -> {
                    if (in.equals("white")) out.collect(in);
                    else if (in.equals("black")) {
                        out.collect(in);
                        out.collect(in);
                    }
                })
                // 泛型擦除
                // Collector<T> out ---> Collector<Object> out
                // 提供输出的类型信息
                .returns(Types.STRING)
                .print("lambda");

        env
                .fromElements(1)
                .map(r -> Tuple2.of(r, r))
                // Tuple2<T0, T1> ---> Tuple2<Object, Object>
                .returns(Types.TUPLE(Types.INT, Types.INT))
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class MyFlatMap implements FlatMapFunction<String, String> {
        @Override
        public void flatMap(String s, Collector<String> collector) throws Exception {
            if (s.equals("white")) collector.collect(s);
            else if (s.equals("black")) {
                collector.collect(s);
                collector.collect(s);
            }
        }
    }
}
